package client

import (
	"encoding/json"
	"fmt"
	"github.com/IBM/sarama"
	"github.com/zeromicro/go-zero/core/logx"
	"go.uber.org/zap"
	"listenboos-common/kafka/model"
	model_redis "listenboos-common/redis/model"
)

var KafKaClient = Init()

// MsgChatTransferClient  kafka客户端接口
type MsgChatTransferClient interface {
	Push(msg *model.AlbumInfo) error
}
type KafkaProducer struct {
	producer sarama.SyncProducer
}

func (k *KafkaProducer) Push(msg *model.AlbumInfo) error {

	body, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	_, _, err = k.producer.SendMessage(&sarama.ProducerMessage{
		Topic: "albumInfo",
		Value: sarama.ByteEncoder(string(body)),
	})
	return err
}
func Init() sarama.Client {
	saramaCfg := sarama.NewConfig()
	saramaCfg.Producer.Return.Successes = true
	client, err := sarama.NewClient([]string{"127.0.0.1:9094"}, saramaCfg)
	if err != nil {
		zap.L().Error("kafka consumer InitKafkaClient:", zap.Error(err))
		return nil
	}
	return client
}
func PushV2ToSuggest(msg *model.SuggestInfo) error {
	saramaCfg := sarama.NewConfig()
	saramaCfg.Producer.Return.Successes = true
	client, err := sarama.NewClient([]string{"127.0.0.1:9094"}, saramaCfg)
	if err != nil {
		zap.L().Error("kafka consumer InitKafkaClient:", zap.Error(err))
		return nil
	}
	res, err := sarama.NewSyncProducerFromClient(client)
	if err != nil {
		panic(err)
	}
	body, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	_, _, err = res.SendMessage(&sarama.ProducerMessage{
		Topic: "suggest_info",
		Value: sarama.ByteEncoder(string(body)),
	})
	if err != nil {
		logx.Errorf("发送msg失败,error:%v", err)
		return err
	}
	fmt.Println("已经成功发送消息")
	return nil
}
func PushV1(msg *model.AlbumInfo) error {
	saramaCfg := sarama.NewConfig()
	saramaCfg.Producer.Return.Successes = true
	client, err := sarama.NewClient([]string{"127.0.0.1:9094"}, saramaCfg)
	if err != nil {
		zap.L().Error("kafka consumer InitKafkaClient:", zap.Error(err))
		return nil
	}
	res, err := sarama.NewSyncProducerFromClient(client)
	if err != nil {
		panic(err)
	}
	body, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	_, _, err = res.SendMessage(&sarama.ProducerMessage{
		Topic: "albumInfo",
		Value: sarama.ByteEncoder(string(body)),
	})
	if err != nil {
		logx.Errorf("发送msg失败,error:%v", err)
		return err
	}
	fmt.Println("已经成功发送消息")
	return nil
}
func PushToProcess(msg *model_redis.KafkaAlbumAndTrackPlaySum) error {
	saramaCfg := sarama.NewConfig()
	saramaCfg.Producer.Return.Successes = true
	client, err := sarama.NewClient([]string{"127.0.0.1:9094"}, saramaCfg)
	if err != nil {
		zap.L().Error("kafka consumer InitKafkaClient:", zap.Error(err))
		return nil
	}
	res, err := sarama.NewSyncProducerFromClient(client)
	if err != nil {
		panic(err)
	}
	body, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	_, _, err = res.SendMessage(&sarama.ProducerMessage{
		Topic: "process",
		Value: sarama.ByteEncoder(string(body)),
	})
	if err != nil {
		logx.Errorf("发送msg失败,error:%v", err)
		return err
	}
	fmt.Println("已经成功发送消息")
	return nil
}
func PushCollect(client sarama.Client, msg *model.TrackModel) error {
	res, err := sarama.NewSyncProducerFromClient(client)
	if err != nil {
		panic(err)
	}
	body, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	_, _, err = res.SendMessage(&sarama.ProducerMessage{
		Topic: "user_collect",
		Value: sarama.ByteEncoder(string(body)),
	})
	if err != nil {
		logx.Errorf("发送msg失败,error:%v", err)
		return err
	}
	fmt.Println("已经成功发送消息")
	return nil
}
func Push(client sarama.Client, msg *model.AlbumInfo) error {
	res, err := sarama.NewSyncProducerFromClient(client)
	if err != nil {
		panic(err)
	}
	body, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	_, _, err = res.SendMessage(&sarama.ProducerMessage{
		Topic: "albumInfo",
		Value: sarama.ByteEncoder(string(body)),
	})
	if err != nil {
		logx.Errorf("发送msg失败,error:%v", err)
		return err
	}
	fmt.Println("已经成功发送消息")
	return nil
}
func NewKafkaProducer(pc sarama.SyncProducer) MsgChatTransferClient {
	return &KafkaProducer{
		producer: pc,
	}
}
func NewSyncProducer(client sarama.Client) sarama.SyncProducer {
	res, err := sarama.NewSyncProducerFromClient(client)
	if err != nil {
		panic(err)
	}
	return res
}
